package com.lagou.hbase.processor;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

public class HbaseClientDemo {
    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }


    // 创建表方法
    private void createRelationTable() throws IOException {
        final Admin admin = conn.getAdmin();
        final HTableDescriptor relation = new HTableDescriptor(TableName.valueOf("user_r"));
        final HColumnDescriptor cf = new HColumnDescriptor("friends");
        final HTableDescriptor hTableDescriptor = relation.addFamily(cf);
        admin.createTable(hTableDescriptor);

        System.out.println("user_r表创建成功！！");
        admin.close();
    }

    //初始化部分数据
    private void initRelationData() throws IOException {
        final Table relation = conn.getTable(TableName.valueOf("user_r"));
        //插入uid1用户
        final Put uid1 = new Put(Bytes.toBytes("uid1"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        uid1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid4"), Bytes.toBytes("uid4"));
        //插入uid2用户
        final Put uid2 = new Put(Bytes.toBytes("uid2"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        uid2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid4"), Bytes.toBytes("uid4"));
        //准备list集合
        final ArrayList<Put> puts = new ArrayList<Put>();
        puts.add(uid1);
        puts.add(uid2);
        //执行写入
        relation.put(puts);
        System.out.println("数据初始化成功！！");
    }


    //某个用户其中一个好友，
    private void deleteFriends(String uid, String friend) throws IOException {
        /*
        1 需要根据传入用户删除其列族中对应好友的列
        2 其好友的列族对应需要删除该用户的列
         */
        final Table relation = conn.getTable(TableName.valueOf("user_r"));
        final Delete delete = new Delete(Bytes.toBytes(uid));
        delete.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(friend));//最好使用协处理器实现

        relation.delete(delete);

    }

    @Test
    public void testCreate() throws IOException {
        createRelationTable();//创建表
        initRelationData();//初始化数据

    }

    @Test
    public void testDelete() throws IOException{
        //假设uid1用户删除了uid2这个好友，需要执行删除操作
        deleteFriends("uid1", "uid2");
    }


    //释放连接
    @After
    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
